/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.transport.threadpool;

import java.nio.ByteBuffer;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;

import cgl.narada.transport.TransportDebugFlags;
import cgl.narada.transport.TransportHandler;
import cgl.narada.util.ByteUtilities;

/**
   Listens for PoolTCP packets arriving at a socket and performs various operations
   based on situations arising during communication. These received packets 
   are then encapsulated into a specialized data structure and forwarded to the
   transport handler implenetation residing in the transport layer.
   
   @author Shrideep Pallickara
   $Revision$
   $Date$
*/

public class PoolTCPLinkReceiver implements TransportDebugFlags {
  
  private SocketChannel sc;
  private ByteBuffer buffer;
  private Selector readSelector;
  private int BUFFER_SIZE=8192;

  private int packetTracker=0;
  private boolean zeroLengthRead = false;
  private long readingDelayStart;
  private int intraPacketTracker=0;

  private PoolTCPLink pooltcpLink;

  private String moduleName = "PoolTCPLinkReceiver: ";
  private TransportHandler transportHandler;
  private String linkId;
  
  
  private int numToRead=0, readSoFar=0;
  private byte[] dataRecvd;
  private int numToReadOffset=0;
  private byte[] numToReadBytes;
  private int lastSizeRead = 0;
  private int tracker=0;
  private int oldTracker=0;
  private boolean keepLooping = true;

  public PoolTCPLinkReceiver(SocketChannel sc, PoolTCPLink pooltcpLink)
    throws Exception {
    this.pooltcpLink = pooltcpLink;
    this.sc = sc;
        
    buffer = ByteBuffer.allocateDirect(BUFFER_SIZE*8);
    sc.configureBlocking(false);
    //readSelector = Selector.open();
    //sc.register(readSelector, SelectionKey.OP_READ);
    transportHandler = pooltcpLink.getTransportHandler();
    if (transportHandler == null ) {
      System.out.println(moduleName + "TransportHandler is null");
    }

    numToReadBytes = new byte[4];
  }
  
  
  public void setLinkId(String linkId) {
    if (linkId == null) {
      System.out.println(moduleName + "Specified Link ID is null");
    }
    this.linkId = linkId;
  }

  public void releaseResources() {
    if (PoolTCPLinkReceiver_Debug)
      System.out.println(moduleName + "Closing " + sc );
    try {
      sc.close();
    } catch (Exception ioe) {
      System.out.println(moduleName + "Problem while releasing resources" +
			 ioe);
    }
  }
  
  
  public void readDataFromSocket() throws Exception { 
    if (PoolTCPLinkReceiver_Debug) 
      System.out.println("Receiving ..");
        
    int count;
    buffer.clear();
    
    while ( (count = sc.read(buffer)) > 0) {
      intraPacketTracker += count;
      tracker += count;
      

      if (PoolTCPLinkReceiver_Debug) {
	System.out.println("Read ..." + tracker);
	System.out.println("intraPacketTracker=" +  intraPacketTracker);
      }
      buffer.flip();
      processBuffer(count);
      
      /*if ( (lastSizeRead+4) > buffer.capacity()) {
	System.out.println(moduleName + "Reallocating buffer \n \n");
	buffer = ByteBuffer.allocateDirect(lastSizeRead+4);
	} */
    }
    
    if (count < 0) {
      System.out.println(moduleName + "Proceeding to close channel");
      sc.close();
      keepLooping = false;
    }
    
  }/* end readFromSocket */  
  
  private void processBuffer(int count) {
    //if (PoolTCPLinkReceiver_Debug) {
      System.out.println("Processing buffer on link " + linkId + 
			 " ... size=" + count);
      //}
    int advance =0;
    if (numToRead == 0) {
      if (numToReadOffset == 0) {
	if (count >= 4) {
	  numToRead = buffer.getInt();
	  advance += 4;
	  allocateDataReceived(numToRead);
	} else {
	  buffer.get(numToReadBytes, 0, count);
	  numToReadOffset = count;
	  System.out.println("numToReadOffset = " + numToReadOffset); 
	  advance += count;
	}
      }	else {
	if ( (numToReadOffset + count) >= 4) {
	  System.out.println("numToReadOffset = " + numToReadOffset); 
	  buffer.get(numToReadBytes, numToReadOffset, (4-numToReadOffset));
	  numToRead = ByteUtilities.getInt(numToReadBytes);
	  allocateDataReceived(numToRead);
	  advance += (4-numToReadOffset);
	  numToReadOffset = 0;
	} else {
	  buffer.get(numToReadBytes, numToReadOffset, count);
	  numToRead = ByteUtilities.getInt(numToReadBytes);
	  numToReadOffset += count;
	  System.out.println("numToReadOffset = " + numToReadOffset); 
	  advance +=count;
	}       
      } /** numToRead Offset */
    } /** numToRead */
    
    if (advance == count) {
      return;
    }
    
    int availableToRead = count - advance;
    if (PoolTCPLinkReceiver_Debug) {
      System.out.println("availableToRead=" + availableToRead + 
			 ", numToRead=" + numToRead + ", readSoFar=" + 
			 readSoFar + ", totalReceived=" + tracker);
    }
    if (readSoFar == 0) {
      if (availableToRead >= numToRead) {
	buffer.get(dataRecvd, 0, numToRead);
	advance += numToRead;
	//if (PoolTCPLinkReceiver_Debug) {
	  System.out.println("Reporting ["+ numToRead + 
			     "] to TransportHandler in one read");
	  //}
	transportHandler.dataReceived(dataRecvd, linkId);
	numToRead = 0;
	readSoFar =0;
	intraPacketTracker -= (numToRead+4);
	if ( (count-advance) > 0) 
	  processBuffer(count-advance);
      } else {
	buffer.get(dataRecvd, 0, availableToRead);
	readSoFar = availableToRead;
	return;
      }
    } else {
      if ((availableToRead + readSoFar) >=  numToRead) {
	buffer.get(dataRecvd, readSoFar, (numToRead - readSoFar));
	advance += (numToRead - readSoFar);
	readSoFar += (numToRead - readSoFar);
	if (PoolTCPLinkReceiver_Debug) { 
	  System.out.println("Reporting ["+ numToRead + 
			     "] to TransportHandler greater reads");
	}
	transportHandler.dataReceived(dataRecvd, linkId);
	intraPacketTracker -= (numToRead+4);
	numToRead =0;
	readSoFar =0;
	if ( (count-advance) > 0) 
	  processBuffer(count-advance);
      } else {
	buffer.get(dataRecvd, readSoFar, availableToRead);
	readSoFar += availableToRead;
      }
    }
  }

  private void allocateDataReceived(int size) {
    if (size > 100000) { 
      System.out.println("Number of bytes to read = [" + size + "] on link " +
			 linkId);
    }
    
    dataRecvd = new byte[size];
  }
  
  
}


